package com.example.influxdb01.controller;

import com.example.influxdb01.pojo.PackageInfo;
import com.example.influxdb01.service.PackageInfoWriteService;
import com.example.influxdb01.utils.InfluxdbUtils;
import com.sun.javaws.jnl.ResourcesDesc;
import org.influxdb.InfluxDB;
import org.influxdb.dto.BatchPoints;
import org.influxdb.dto.Point;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.lang.reflect.Field;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @author yihao
 * @description
 * @time 2022/04/26 09:26:36
 */
@RestController
public class IndexController {
    private static Logger LOGGER = LoggerFactory.getLogger(IndexController.class);

    @Autowired
    private InfluxDB influxDB;

    @Autowired
    private InfluxdbUtils influxdbUtils;

    //measurement
    private final String measurement = "cpu_load_short";

    @Value("${spring.influx.database}")
    private String database;

    /**
     * 批量插入第一种方式
     */
    @GetMapping("/index")
    public void insert(){
        List<String> lines = new ArrayList<String>();
        Point point = null;
        for(int i=0;i<50;i++){
            point = Point.measurement(measurement)
                    .tag("host", "server0" + i)
                    .tag("region","us-west")
                    .addField("value", (float) i)
                    .build();
            lines.add(point.lineProtocol());
        }
        //写入
        influxDB.write(lines);
        LOGGER.info("写入完成");


    }

    /**
     * 批量插入第二种方式
     */
    @GetMapping("/index1")
    public void batchInsert(){
        BatchPoints batchPoints = BatchPoints
                .database(database)
                .consistency(InfluxDB.ConsistencyLevel.ALL)
                .build();
        //遍历sqlserver获取数据
        for(int i=0;i<50;i++){
            //创建单条数据对象——表名
            Point point = Point.measurement(measurement)
                    //tag属性——只能存储String类型
                    .tag("deviceId", "sensor" + i)
                    .addField("temp", 3)
                    .addField("voltage", 145+i)
                    .addField("A1", "4i")
                    .addField("A2", "4i").build();
            //将单条数据存储到集合中
            batchPoints.point(point);
        }
        //批量插入
        influxDB.write(batchPoints);
    }


    /**
     * 获取数据
     */
    @GetMapping("/datas")
    public void datas(@RequestParam Integer page){
        int pageSize = 10;
        // InfluxDB支持分页查询,因此可以设置分页查询条件
        String pageQuery = " LIMIT " + pageSize + " OFFSET " + (page - 1) * pageSize;

        String queryCondition = "";
        // 此处查询所有内容,如果
        String queryCmd = "SELECT * FROM cpu_load_short"
                // 查询指定设备下的日志信息
                // 要指定从 RetentionPolicyName.measurement中查询指定数据,默认的策略可以不加；
                // + 策略name + "." + measurement
                // 添加查询条件(注意查询条件选择tag值,选择field数值会严重拖慢查询速度)
                + queryCondition
                // 查询结果需要按照时间排序
                + " ORDER BY time DESC"
                // 添加分页查询条件
                + pageQuery;

        List<Object> sensorList = influxdbUtils.fetchRecords(queryCmd);
        LOGGER.info("query result => {}", sensorList);
    }

    /**
     * 使用指定的pojo来接获取数据
     */
    @GetMapping("/datas01")
    public void datas1(@RequestParam Integer page){
        //目前没有使用
        int pageSize = 8000;
        // InfluxDB支持分页查询,因此可以设置分页查询条件
        String pageQuery = " LIMIT " + pageSize + " OFFSET " + (page - 1) * pageSize;

        //拿到从配置文件中初始化好的查询时间
        String queryStartTime = PackageInfoWriteService.getQueryStartTime();
        String queryEndTime = PackageInfoWriteService.getQueryeEndTime();

        //获取配置中的包和相应的参数集合
        Map<String, Integer> packagesWithParamCount = PackageInfoWriteService.getPackagesWithParamCount();
        String[] params = PackageInfoWriteService.getParams();

        //针对每一个包做一次查询，生成一个csv文件
        Iterator<String> iterator = packagesWithParamCount.keySet().iterator();
        int paramsCount = 0;

        //累计每个包的参数数量
        int index = 0;

        //获取已经初始化好了的转置后结果集的文件头
        Map<String, String[]> resultFileHeaders = PackageInfoWriteService.getResultFileHeaders();
        Iterator<String> iterator1 = resultFileHeaders.keySet().iterator();

        while (iterator.hasNext()) {

            String packageName = iterator.next();
            StringBuilder queryCondition =
                    //查询条件需要用双引号，条件值需要用单引号
                    new StringBuilder("SELECT * FROM \"ResultTM\" where \"PackageName\" = \'" + packageName+"\'  AND time >= \'"+queryStartTime+"\' AND time <= \'"+queryEndTime+"\' AND (");
            paramsCount += packagesWithParamCount.get(packageName);
            for (int i = index; i < paramsCount; i++) {
                queryCondition.append("\"ParaName\" = \'" + params[i] + "\'");
                if (i < paramsCount-1) {
                    queryCondition.append(" or ");
                }
            }

            //把当前包的参数数量进行累积，下一次读取就从第二个包的第一个参数开始了
            index += paramsCount;
            queryCondition.append(")");
            queryCondition.append(" ORDER BY time DESC");
            //queryCondition.append(pageQuery);
            LOGGER.info("查询语句为：{}",queryCondition.toString());

            List<PackageInfo> packageInfoList = influxdbUtils.fetchResults(queryCondition.toString(), PackageInfo.class);
            packageInfoList.forEach( packageInfo -> {
                LOGGER.debug("query packageInfo => {}", packageInfo);
            });

            //把pojo对象再转化成字符串数组 用于第三方csv API写入文件
            List<String[]> result = convertPojoToStringArray(packageInfoList);

            PackageInfoWriteService packageInfoService = new PackageInfoWriteService();
            //从配置中获取 是否合并输出
            if (PackageInfoWriteService.isCompound()) {
                packageInfoService.wirte(result, PackageInfoWriteService.getCompoundFileName(), PackageInfoWriteService.getFileHeader());
            }else{
                packageInfoService.wirte(result,packageName,PackageInfoWriteService.getFileHeader());
            }


            if (iterator1.hasNext()) {
                LinkedList<String[]> resultFiles = new LinkedList<>();
                String currentPackageName = iterator1.next();
                String[] resultFormat = resultFileHeaders.get(currentPackageName);

                LOGGER.info("转置后的packageName :{} 文件头：{}", currentPackageName,Arrays.toString(resultFormat));

                //转置结果集
                transposeResult(packageInfoList, resultFiles, resultFormat);

                //写csv数据
                packageInfoService.wirte(resultFiles,packageName+"modify",resultFormat);

            }

        }
    }

    /**
     * @author yihao
     * 把结果集转置，通过文件头数组匹配数据行中参数的名称，来给转置结果集的元素数组中插入相应位置的数据
     * @param packageInfoList 原始pojo结果集
     * @param resultFiles 转置结果集
     * @param resultFormat 转置结果集文件头
     */
    private void transposeResult(List<PackageInfo> packageInfoList, LinkedList<String[]> resultFiles, String[] resultFormat) {

        //由于按时间排序后，其参数序号也有顺序，所以每次跳 文件头中字段数量减1（没有time）即构成一组，这一组的值就可以赋值给一个转置结果集中的元素
        for (int i = 0; i < packageInfoList.size(); i+=(resultFormat.length -1)) {
            String[] resultModify = new String[resultFormat.length];
            resultModify[0] = packageInfoList.get(i).getTime();
            for (int j = 0; j < resultFormat.length -1 ; j++) {
                if (i + j < packageInfoList.size()) {
                    //这里使用i+j 目的是保证外层循环resultFormat.length -1跳一次，而内层循环要每次都跳
                    int formatIndex = getIndex(resultFormat, packageInfoList.get(i+j).getParaName());
                    resultModify[formatIndex] = packageInfoList.get(i+j).getResultPhy().toString();
                }

            }
            resultFiles.add(resultModify);
        }
    }

    /**
     * @author yihao
     * 把pojo集合转成csvAPI可以使用的String[]集合
     * @param packageInfoList 原始的pojo集合
     * @return
     */
    private List<String[]> convertPojoToStringArray(List<PackageInfo> packageInfoList) {
        Class clazz = PackageInfo.class;
        Field[] declaredFields = clazz.getDeclaredFields();
        return packageInfoList.stream().map(packageInfo -> {
            String[] packageInfoStr = new String[declaredFields.length];
            try {
                //这里进行了动态类型判断，同时动态获取getter来给String[]元素赋值
                for (int i = 0; i < declaredFields.length; i++) {
                    if (declaredFields[i].getType().equals(Double.class)) {
                        packageInfoStr[i] = ((Double)clazz.getDeclaredMethod("get"+captureWords(declaredFields[i].getName())).invoke(packageInfo)).toString();
                        continue;
                    }
                    packageInfoStr[i] = (String)clazz.getDeclaredMethod("get"+captureWords(declaredFields[i].getName())).invoke(packageInfo);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
            LOGGER.debug("pojo:{}", Arrays.toString(packageInfoStr));
            return packageInfoStr;
        }).collect(Collectors.toList());
    }

    /**
     * @author yihao
     * 单词首字母大写，目的是拼接get 反射调用getter
     * @param words
     * @return
     */
    private String captureWords(String words) {
        words = words.substring(0, 1).toUpperCase() + words.substring(1);
        return  words;
    }

    /**
     * @author yihao
     * 获取数组元素在数组中的下标，
     * @param strings
     * @param target
     * @return
     */
    private int getIndex(String[] strings,String target) {
        for (int i = 0; i < strings.length; i++) {
            if (strings[i].equals(target)) {
                return i;
            }
        }
        return -1;
    }
}

